package utils

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

import java.util.Properties

/**
 * 实现producer
 */
object KafkaProducer {
  def send(topic: String, msg: String) {
    val prop = new Properties
    // 指定请求的kafka集群列表
    prop.put("bootstrap.servers", "192.168.81.133:9092") // 指定响应方式
    //prop.put("acks", "0")
    prop.put("acks", "all")
    // 请求失败重试次数
    //prop.put("retries", "3")
    // 指定key的序列化方式, key是用于存放数据对应的offset
    prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    // 指定value的序列化方式
    prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    // 配置超时时间
    prop.put("request.timeout.ms", "60000")

    // 得到生产者的实例
    val producer = new KafkaProducer[String, String](prop)
    val rmd: RecordMetadata = producer.send(new ProducerRecord[String, String](topic, msg)).get()

    producer.close()
  }
}
